extern crate threadpool;

// use serde::{Serialize, Deserialize};
// use serde_json::{json,Value};
use colored::Colorize;
use log::{error, info};
use serde_json::json;
use std::sync::Mutex;
use url::Url;

use std::fs;
use std::io::Read;
use std::path::Path;
use std::sync::mpsc::channel;
use threadpool::ThreadPool;

use super::utils::{self, now, to_safe, SafeSender, StringSplit};

use super::backends::Nodes;
use super::backends::TmpData;
use super::check::SystemInfoAndBuild;
use super::client::json_to;
use super::config::get_config_password;
use super::encrypt::Encryptor;
use super::routes::Parameter;
use super::utils::StringFile;
use super::utils::OutputObject;

const BRD_SEP: &str = ":::";
const ARG_SEP: &str = "@\n";

lazy_static! {
    static ref BRD_RUNER: Mutex<ThreadPool> = Mutex::new(ThreadPool::new(4));
    static ref CRYPTOR: Mutex<Encryptor> = {
        let pwd = get_config_password();
        let map = Encryptor::new(&pwd);
        Mutex::new(map)
    };
    static ref BRD_CENTR: Mutex<Broadcaster> = Mutex::new(Broadcaster::brd_initialization());
}

pub struct Broadcaster {
    tx: SafeSender,
    // brd_list: Vec<String>,
}

impl Broadcaster {
    fn brd_initialization() -> Broadcaster {
        let (tx, rx) = channel::<String>();
        // let db_h = Arc::clone(&dbs);
        let pool = BRD_RUNER.lock().expect("unlock threadpool failed");
        pool.execute(move || {
            loop {
                let out = match rx.recv() {
                    Ok(a) => a,
                    _ => break,
                };
                // patch a brd thread
                if out.starts_with("[BRD_SEND]") {
                    continue;
                }
                if out.starts_with("[BRD]") && out.contains(BRD_SEP) {
                    let (_, op) = out.split_once(BRD_SEP);
                    if op == "del" {
                        Nodes::remove_node("[BRD]").expect("remove [BRD] failed");
                    }
                    continue;
                }
                if !out.starts_with("http") || !out.contains(BRD_SEP) {
                    continue;
                }
                let (host, status) = out.split_once(BRD_SEP);
                match Nodes::node(&host, &status) {
                    Ok(_) => {
                        OutputObject::brd_task_to_object(&host);
                        info!("update broadcast: {}:{}", host.blue(), status.green());
                    }
                    _ => {
                        error!("update host status failed ? {} : {}", host, status);
                    }
                }
            }
        });
        Broadcaster { tx: to_safe(tx) }
    }

    pub fn get_sender(&self) -> SafeSender {
        self.tx.clone()
    }

    fn hosts(&self) -> Vec<String> {
        Nodes::nodes()
    }

    pub fn onlines(&self) -> Vec<String> {
        Nodes::nodes_filter("online")
    }

    pub fn get_brd(&self) -> (String, String, String) {
        let s = Nodes::get("[BRD]");
        let out = s.split_twice(BRD_SEP);
        Nodes::remove_node("[BRD]").expect("remove [BRD] failed");
        out
    }

    pub fn remove_brd(&self) {
        Nodes::remove_node("[BRD]").expect("remove [BRD] failed");
    }
}

pub trait BroadcastBatch {
    // add code here
    fn batch_tasks(&self) -> String;
}

impl BroadcastBatch for Vec<String> {
    fn batch_tasks(&self) -> String {
        self.join(ARG_SEP)
    }
}

impl BroadcastBatch for Vec<&str> {
    fn batch_tasks(&self) -> String {
        self.join(ARG_SEP)
    }
}

pub trait BroadcastBatchStr {
    fn batch_from_file(&self) -> Option<String>;
    fn batch_from_file_task(&self) -> Option<String>;
    fn batch_with_sep(&self, sep: &str) -> String;
}

impl<'a> BroadcastBatchStr for &'a str {
    fn batch_from_file_task(&self) -> Option<String> {
        let (pre, f) = self.split_once(":");
        let (_, tasks_file) = f.split_once("@");
        let p = Path::new(tasks_file.trim());
        let mut buf: String = String::new();
        if p.exists() {
            let mut f = match fs::File::open(p) {
                Ok(a) => a,
                Err(e) => {
                    error!("reason : {} {}", self.red(), e);
                    return None;
                }
            };
            match f.read_to_string(&mut buf) {
                Ok(a) => a,
                Err(_) => {
                    return None;
                }
            };
            let lines: Vec<String> = buf
                .split("\n")
                .map(|x| {
                    let s = format!("{}:{}", &pre, x);
                    s
                })
                .collect();
            return Some(lines.batch_tasks());
        } else {
            return None;
        }
    }
    fn batch_from_file(&self) -> Option<String> {
        let p = Path::new(self);
        let mut buf: String = String::new();
        if p.exists() {
            let mut f = match fs::File::open(p) {
                Ok(a) => a,
                Err(e) => {
                    error!("reason : {} {}", self.red(), e);
                    return None;
                }
            };
            match f.read_to_string(&mut buf) {
                Ok(a) => a,
                Err(_) => {
                    return None;
                }
            };
            let lines: Vec<&str> = buf.split("\n").collect();
            return Some(lines.batch_tasks());
        }
        None
    }

    fn batch_with_sep(&self, sep: &str) -> String {
        let lines: Vec<&str> = self.split(sep).collect();
        return lines.batch_tasks();
    }
}

impl BroadcastBatchStr for String {
    fn batch_from_file_task(&self) -> Option<String> {
        self.as_str().batch_from_file_task()
    }

    fn batch_from_file(&self) -> Option<String> {
        let p = Path::new(&self);
        let mut buf: String = String::new();
        if p.exists() {
            let mut f = match fs::File::open(p) {
                Ok(a) => a,
                Err(e) => {
                    error!("reason : {} {}", self.red(), e);
                    return None;
                }
            };
            match f.read_to_string(&mut buf) {
                Ok(a) => a,
                Err(_) => {
                    return None;
                }
            };
            let lines: Vec<&str> = buf.split("\n").collect();
            return Some(lines.batch_tasks());
        }
        None
    }

    fn batch_with_sep(&self, sep: &str) -> String {
        let lines: Vec<&str> = self.split(sep).collect();
        return lines.batch_tasks();
    }
}

pub fn view_brd_ids(id: &str, ids: &Vec<String>) {
    // let brd = BRD_CENTR.lock().expect("unlock brd centry");
    // let tx = brd.get_sender();
    // let mut out = String::new();
    let pool = BRD_RUNER.lock().expect("unlock threadpool failed");
    for uid in ids {
        let (uid, host) = uid.split_once(" || ");

        let raw_id = id.to_string();
        pool.execute(move || {
            // let brd = BRD_CENTR.lock().expect("unlock brd centry");
            let cr = CRYPTOR.lock().unwrap();
            let r_id = raw_id.clone();
            let repl = json_to(
                "view",
                &uid,
                "",
                "/task",
                Some(&host.trim()),
                Some(&cr),
                None,
            );
            if repl.status == 0 && repl.msg.len() > 0 {
                if let Some(mut o) = r_id.uuid_to_object(){
                    let mut  buf = String::new();
                    if let Some(sub_obj) = repl.msg.to_output_object(){
                        buf.push_str(&sub_obj.get("cmd").unwrap());
                        buf.push_str(&"\n".bold().green().to_string());
                        buf.push_str(&sub_obj.get("content").unwrap());
                    }else{
                        buf = repl.msg;
                    }
                    o.insert(&uid, &buf);
                    o.save();
                }
                // utils::append_to_file(&r_id, &uid, &repl.msg.as_bytes());
                let old = TmpData::query(&r_id);
                let old_s = old.as_str().unwrap();
                TmpData::insert(&r_id, json!(&format!("{}{}", old_s, &uid)));
                
            }
        });
        // out.push_str(&format!("{}\n", &uid));
    }
    // out
}

pub fn broadcast(para: &Parameter) -> (i32, String) {
    let brd_op = para.get_op();
    let raw_op = format!("{}", &brd_op[4..brd_op.len() - 1]);

    let brd = BRD_CENTR.lock().expect("unlock brd centry");
    // let tx = brd.get_sender();
    let pool = BRD_RUNER.lock().expect("unlock threadpool failed");
    let data = para.data.clone();
    let tp = para.data_tp.clone();
    let task_id = uuid::Uuid::new_v4().to_string();
    let mut count = 0;
    // let _ = Nodes::node("[BRD]", &vec![raw_op, data, tp].join(BRD_SEP));
    // let (msg_tx, msg_rx) = std::sync::mpsc::channel::<String>();
    let mut hosts: String = String::new();

    if (&data).contains(ARG_SEP) {
        let payloads: Vec<&str> = (&data).split(ARG_SEP).collect();
        let hosts_iter = brd.onlines();
        if hosts_iter.len() == 0 {
            return (2, "no nodes to handle".to_string());
        }
        let mut no = 0;
        for payload in payloads {
            count += 1;
            let host = hosts_iter[no % hosts_iter.len()].to_string();
            let pay = vec![raw_op.clone(), payload.to_string(), tp.clone()].join(BRD_SEP);
            info!("brd task {} -> {}", no, &host);
            hosts.push_str(&format!("{}/{}\n", host, &raw_op));
            let uid = task_id.clone();
            // let msg_tx = msg_tx.clone();
            pool.execute(move || {
                // let op = raw_op.clone();
                // println!("{}", &tp);
                // let msg_tx_s = msg_tx.clone();
                let send_uid = uid.clone();
                let pay_share = pay.clone();
                let brd = BRD_CENTR.lock().expect("unlock brd centry");
                let cr = CRYPTOR.lock().unwrap();
                let (op_uri, data, tp) = pay_share.split_twice(BRD_SEP);

                // let mut payload:String;
                let (op, uri) = op_uri.split_once(":");
                let repl = json_to(
                    &op.trim(),
                    &data,
                    &tp,
                    &uri.trim(),
                    Some(&host.trim()),
                    Some(&cr),
                    None,
                );
                let tx = brd.get_sender();
                let tx_shared = tx.lock().unwrap();
                if repl.status == 0 {
                    let v:serde_json::Value = match serde_json::from_str(&repl.msg){
                        Ok(a) => a,
                        Err(e) => {
                        
                            log::error!("{}",e);
                            // return (1, format!("parse json failed from {}", host));
                            json!("")
                        }
                    };
                    
                    let task_id_c = match v.get("id"){
                        Some(a) => a.as_str().unwrap(),
                        _ => {
                            log::error!("get task id as str failed");
                            
                            ""
                        }
                    };
                    if task_id_c.len() > 0{
                        tx_shared
                        .send(format!(
                            "{} || {} || {}{}{}",
                            &host, &send_uid, task_id_c, BRD_SEP, "wait"
                        ))
                        .expect("send failed");
                    }else{
                        let _ = tx_shared
                        .send(format!(
                            "{} || {} || {}{}{}",
                            &host, &send_uid, &task_id_c, BRD_SEP, "failed try again"
                        ));
                    }
                    

                } else if repl.status == 1 {
                    tx_shared
                        .send(format!("{}{}{}", &host, BRD_SEP, "off"))
                        .expect("send brd failed");
                } else {
                    info!(
                        "{}-{} | {} | reason: {} ",
                        &host.green(),
                        "brd",
                        "send failed!".red(),
                        &repl.msg.red()
                    );
                }
                // msg_tx_s
                //     .send(repl.msg.clone())
                //     .expect("send failed to smg recv");
            });
            no += 1;
        }
    } else {
        // let _ = Nodes::node("[BRD]", &vec![raw_op, para.data.clone(), tp].join(BRD_SEP));
        let pay = vec![raw_op.clone(), para.data.clone(), tp].join(BRD_SEP);
        // info!("ready a brd : {}", para);
        for host in brd.onlines() {
            hosts.push_str(&format!("{}/{}\n", host, &raw_op));
            // let msg_tx = msg_tx.clone();
            let pay_share = pay.clone();
            let uid = task_id.clone();
            count += 1;
            pool.execute(move || {
                // let op3 = op2.clone();

                // let msg_tx_s = msg_tx.clone();
                let send_uid = uid.clone();
                let payload = pay_share.clone();
                let brd = BRD_CENTR.lock().expect("unlock brd centry");
                let cr = CRYPTOR.lock().unwrap();
                let (op_uri, data, tp) = payload.split_twice(BRD_SEP);

                // let mut payload:String;
                let (op, uri) = op_uri.split_once(":");
                info!(
                    "op:{} data:{} data_tp:{} host:{}/{} encrypt:{} brd:{} ",
                    &op, &data, &tp, &host, &uri, "T", "F"
                );
                let repl = json_to(
                    &op.trim(),
                    &data,
                    &tp,
                    &uri.trim(),
                    Some(&host.trim()),
                    Some(&cr),
                    None,
                );
                let tx = brd.get_sender();
                let tx_shared = tx.lock().unwrap();
                if repl.status == 0 {
                    let v: serde_json::Value = serde_json::from_str(&repl.msg).unwrap();
                    let task_id = v
                        .get("id")
                        .expect("no such id")
                        .as_str()
                        .expect("get task id as str failed");
                    tx_shared
                        .send(format!(
                            "{} || {} || {}{}{}",
                            &host, &send_uid, task_id, BRD_SEP, "wait"
                        ))
                        .expect("send failed");
                // msg_tx_s
                //     .send(repl.msg.clone())
                //     .expect("send failed to smg recv");
                // tx_shared
                //     .send(format!("{}-{}{}{}", &host, "brd", BRD_SEP, "send ok!"))
                //     .expect("send brd failed");
                } else if repl.status == 1 {
                    tx_shared
                        .send(format!("{}{}{}", &host, BRD_SEP, "off"))
                        .expect("send brd failed");
                } else {
                    info!(
                        "{}-{} | {} | reason: {}",
                        &host.green(),
                        "brd",
                        "send failed!".red(),
                        &repl.msg.red()
                    );
                }
            });
            // no += 1;
            // msg_tx.clone().send("[END]".to_string());
        }
    }
    let mut back_msg = String::new();
    // let mut cc = 0;
    // loop {
    //     cc += 1;
    //     let s = msg_rx.recv().expect("exepct");
    //     if count == cc {
    //         break;
    //     }
    //     log::info!("ok [{}]", &s);

    //     if cc % 2 == 0 {
    //         back_msg.push_str(&format!("{}\n", &s).black().on_yellow());
    //     } else {
    //         back_msg.push_str(&format!("{}\n", &s).white().on_green());
    //     }
    // }
    back_msg.push_str(&format!("Running[{}]\n{}", count, &hosts));
    utils::output_to_file(&task_id, "".as_bytes());
    TmpData::insert(&task_id, json!(format!("Running[{}]", count)));
    let out = json! ({
        "id" : &task_id,
        "msg": &back_msg,
    })
    .to_string();
    (0, out)
}

pub fn broadcast_ping(host: &str) {
    let brd = BRD_CENTR.lock().expect("unlock brd centry");
    let tx = brd.get_sender();
    let host_shared = host.to_string();
    let pool = BRD_RUNER.lock().expect("unlock threadpool failed");
    pool.execute(move || {
        let cr = CRYPTOR.lock().unwrap();
        let repl = json_to(
            "ping",
            "who",
            "str",
            "/brd",
            Some(&host_shared),
            Some(&cr),
            None,
        );
        if repl.status == 0 && repl.msg == "ok" {
            info!("broadcast suss {}", repl.msg.green());
            let tx_shared = tx.lock().unwrap();
            tx_shared
                .send(format!("{}{}{}", &host_shared, BRD_SEP, "online"))
                .expect("send brd failed");
        } else {
            info!("broadcast fail {}", repl.msg.red());
        }
    });
}

pub trait Broadcast {
    // add code here
    fn connecting(&self) -> bool;
    fn disconnect(&self) -> bool;
}

impl<'a> Broadcast for &'a str {
    fn connecting(&self) -> bool {
        let url = match Url::parse(self.trim()) {
            Ok(u) => u,
            Err(a) => {
                error!("reason : |{}| {}", self.red(), a);
                return false;
            }
        };
        if let Some(host) = url.host_str() {
            let target_url: String;
            if let Some(port) = url.port() {
                target_url = format!("{}://{}:{}", url.scheme(), host, port);
            } else {
                target_url = format!("{}://{}", url.scheme(), host);
            }

            broadcast_ping(&target_url);
        };
        true
    }

    fn disconnect(&self) -> bool {
        let url = match Url::parse(self.trim()) {
            Ok(u) => u,
            Err(a) => {
                error!("reason : |{}| {}", self.red(), a);
                return false;
            }
        };
        if let Some(host) = url.host_str() {
            let target_url: String;
            if let Some(port) = url.port() {
                target_url = format!("{}://{}:{}", url.scheme(), host, port);
            } else {
                target_url = format!("{}://{}", url.scheme(), host);
            }

            match Nodes::remove_node(&target_url) {
                Ok(_) => {
                    info!("remove : {}", self.trim().green());
                }
                Err(_) => {
                    error!("!remove :{}", self.trim().red());
                }
            }
        }
        true
    }
}

pub fn onlines() -> Vec<String>{
    Nodes::nodes_filter("online")
}

pub fn brd_protocol(para: &Parameter, _: &SystemInfoAndBuild) -> (i32, String) {
    let help = r#"
ping =>  to make sure if host ok.(alway used by program)
ls =>  to see all host's status.
ask => to ask taskid's output.
host => to handle host , data shoutld like 
        {
            op: "host",
            data: "http://localhost, 192.168.1.1:8080, ",
            data_tp: "add/del/"
        }
brd[other:uri] => brd to hosts 
        exam:    
        {
            op: "brd[ls:/db]",
            data: "some normal data",
            data_tp: "some normal tp"
        }
        will brd to all hosts ->   
            $host/db in ..hosts
            {
                op: "ls",
                data: "some normal data",
                data_tp: "some normal tp"
            }
    "#;
    let (op, data, tp) = para.get_args_str();
    let default_back = (1, format!("{}\n{}", help, now()));
    // let default_back_err_host = (1, "host must sep with ',' ".to_string());
    let default_back_err_connect = (1, "connect failed".to_string());

    // brd to all hosts
    if op.starts_with("brd[") && op.ends_with("]") && op.contains(":/") {
        return broadcast(para);
    }
    match op {
        "ping" => (0, "ok".to_string()),
        "ls" => (0, Nodes::status()),
        "ask" => {
            // Nodes::g
            let send_ids = Nodes::get_task_ids(&data);
            view_brd_ids(&data, &send_ids);
            (0, send_ids.join("\n"))
        }
        "host" => {
            if !data.contains(",") {
                if data.connecting() {
                    return (0, data.to_string());
                } else {
                    return default_back_err_connect;
                }
            }
            match tp {
                "" => {
                    let hosts: Vec<String> = Nodes::nodes();
                    (0, hosts.join("\n"))
                }
                "add" => {
                    let hosts: Vec<String> = data
                        .split(",")
                        .filter(|may_host| may_host.connecting())
                        .map(|host| host.to_string())
                        .collect();
                    (0, hosts.join("\n"))
                }
                "del" => {
                    let hosts: Vec<String> = data
                        .split(",")
                        .filter(|may_host| may_host.disconnect())
                        .map(|host| host.to_string())
                        .collect();
                    (0, hosts.join("\n"))
                }
                _ => default_back,
            }
        }
        _ => default_back,
    }
}
